package com.catmiao.rdd.operate.transform;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.storage.StorageLevel;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * @author ChengMiao
 * @title: Transfer_01_Map
 * @projectName spark_study
 * @description: TODO
 * @date 2024/11/25 16:27
 */
public class Transfer_19_persistent {

    public static void main(String[] args) throws InterruptedException {

        final SparkConf conf = new SparkConf();
        conf.setAppName("appName");
        conf.setMaster("local[*]");

        final JavaSparkContext jsc = new JavaSparkContext(conf);


        List<Tuple2<String,Integer>> datas = new ArrayList<>();
        datas.add(new Tuple2<String, Integer>("a",1));
        datas.add(new Tuple2<String, Integer>("a",2));
        datas.add(new Tuple2<String, Integer>("a",3));
        datas.add(new Tuple2<String, Integer>("a",4));

        JavaRDD<Tuple2<String, Integer>> rdd = jsc.parallelize(datas, 2);

        JavaPairRDD<String, Integer> mapRdd = rdd.mapToPair(
                kv -> {
                    System.out.println("111111111111111");
                    return kv;
                }
        );


        // 持久化
        /**
         * 在 RDD重复之间使用持久化操作
         */
        mapRdd.cache();
//        mapRdd.persist(StorageLevel.MEMORY_ONLY());
        mapRdd.persist(StorageLevel.MEMORY_AND_DISK());

        JavaPairRDD<String, Integer> wdRdd = mapRdd.reduceByKey(Integer::sum);
        wdRdd.collect();

        System.out.println("1计算完毕");
        System.out.println("##############################");
        JavaPairRDD<String, Iterable<Integer>> groupRdd = mapRdd.groupByKey();
        groupRdd.collect();

        System.out.println("2计算完毕");


        jsc.close();


    }
}




